package com.seaboxdata.dao;

import com.seaboxdata.entity.OldStrategyConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.time.DateUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @author
 * @create 2020-12-29 16:49
 **/
@Component
@Slf4j
public class HiveDao {
    private static String driverName = "org.apache.hive.jdbc.HiveDriver";

    @Value("${hive.url}")
    private String url;

    @Value("${hive.keytab.path}")
    private String keyTabPath;

    @Value("${hive.principal}")
    private String principal;

    @Value("${krb5.path}")
    private String krb5Path;

    @Value("${hive.database}")
    private String database;

    private Connection connection;

    public Connection getConnection() throws Exception {
        if (connection == null) {
            if (System.getProperty("os.name").toLowerCase().startsWith("win")) {
                this.keyTabPath = "F:/dfjx/keytab/hive.service.keytab";
                this.krb5Path = "F:/dfjx/keytab/krb5.conf";
            }
            System.setProperty("java.security.krb5.conf", krb5Path);
            System.setProperty("sun.security.krb5.debug", "false");
            Configuration conf = new Configuration();
            conf.set("hadoop.security.authentication", "Kerberos");
            UserGroupInformation.setConfiguration(conf);
            conf.set("keytab.file", keyTabPath);
            conf.set("kerberos.principal", principal);
            UserGroupInformation.loginUserFromKeytab(principal, keyTabPath);
            this.connection = DriverManager.getConnection(url);
        }
        return connection;
    }

    public void dataAge(OldStrategyConfig config) {
        Connection connection = null;
        try {
            connection = getConnection();
            Statement statement = connection.createStatement();
            ResultSet resultSet = statement.executeQuery("select data_dt_iso from " + config.getDataPath() + " order by data_dt_iso desc limit 1");
            String latestPartition = null;
            String earliestPartition = null;
            String earliestPartitionInfo = null;
            if (resultSet.next()) {
                //获取最近一次有数据的分区
                latestPartition = resultSet.getString("data_dt_iso");
                //获取最早的分区，得到的是升序的,取最早的分区
                ResultSet partitionsSet = statement.executeQuery("show partitions " + config.getDataPath());
                if (partitionsSet.next()) {
                    earliestPartitionInfo = partitionsSet.getString("partition");
                    earliestPartition = earliestPartitionInfo.substring(earliestPartitionInfo.length() - 10, earliestPartitionInfo.length());
                }
                //循环删除过期数据,获得删除周期
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                Date latestDate = sdf.parse(latestPartition);
                Date earliestDate = sdf.parse(earliestPartition);
                Date deleteDate = DateUtils.addDays(latestDate, -Integer.valueOf(config.getPeriod()));
                if (deleteDate.compareTo(earliestDate) >= 0) {
                    //最早的时间早,需要清理数据
                    Date currentDate = earliestDate;
                    while (currentDate.compareTo(deleteDate) <= 0) {
                        //删除当天数据
                        statement.execute("alter table " + config.getDataPath() + "drop partition (data_dt_iso=" + sdf.format(currentDate) + ")");
                        currentDate = DateUtils.addDays(currentDate, 1);
                    }
                } else {
                    log.info("no partition needs to delete,tableName:" + config.getDataPath());
                }
            }
        } catch (Exception e) {
            log.error("data age error");
            e.printStackTrace();
        }
    }
}
